use std::future::{join, Future};
use std::io::{Read, Write};
use std::net::TcpStream;
use std::pin::Pin;
use std::process::Output;
use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::Duration;

use futures::executor::block_on;
use futures::future::BoxFuture;
use futures::future::MaybeDone::Future as OtherFuture;
use futures::task::{waker_ref, ArcWake};
use futures::FutureExt;

use crate::gen_title;

pub fn ch4_21_1() {
    gen_title("ch4_21_1", 18);

    fn download(url: &str) {
        thread::sleep(Duration::from_millis(10));
        println!("{}", url);
    }

    let thread_one = thread::spawn(|| download("https://baidu.com"));
    let thread_two = thread::spawn(|| download("https://taobao.com"));

    thread_one.join().expect("thread one panicked");
    thread_two.join().expect("thread two panicked");
}

pub fn ch4_21_2() {
    gen_title("ch4_21_2", 18);

    async fn download(url: &str) {
        tokio::time::sleep(Duration::from_millis(10)).await;
        println!("{}", url);
    }

    let future_one = download("https://www.foo.com");
    let future_two = download("https://www.bar.com");

    let future = join!(future_one, future_two);
    block_on(future);
}

pub async fn ch4_21_3() {
    gen_title("ch4_21_3", 18);

    async fn hello_world() {
        hello_cat().await;
        println!("hello world")
    }

    pub async fn hello_cat() {
        println!("hello cat");
    }

    hello_world().await;
}

pub fn ch4_21_4() {
    gen_title("ch4_21_4", 18);

    struct Song {
        author: String,
        name: String,
    }

    async fn learn_song() -> Song {
        Song {
            author: "周杰伦".to_string(),
            name: String::from("《菊花台》"),
        }
    }

    async fn sing_song(song: Song) {
        println!(
            "给大家献上一首{}的{} ~ {}",
            song.author, song.name, "菊花残，吗，满地伤~ ~"
        );
    }

    async fn dance() {
        println!("情到深处，身体不由自主的动了起来~ ~");
    }

    let song = block_on(learn_song());
    block_on(sing_song(song));
    block_on(dance());
}

pub async fn ch4_21_5() {
    gen_title("ch4_21_5", 18);

    struct Song {
        author: String,
        name: String,
    }

    async fn learn_song() -> Song {
        Song {
            author: "周杰伦".to_string(),
            name: String::from("《菊花台》"),
        }
    }

    async fn sing_song(song: Song) {
        println!(
            "给大家献上一首{}的{} ~ {}",
            song.author, song.name, "菊花残，吗，满地伤~ ~"
        );
    }

    async fn dance() {
        println!("情到深处，身体不由自主的动了起来~ ~");
    }

    async fn learn_and_sing() {
        let song = learn_song().await;

        sing_song(song).await;
    }

    let f1 = learn_and_sing();
    let f2 = dance();
    futures::join!(f1, f2);
}

pub fn ch4_21_6() {
    gen_title("ch4_21_6", 18);

    trait SimpleFuture {
        type Output;
        fn poll(&mut self, wake: fn()) -> Poll<Self::Output>;
    }

    #[derive(Debug)]
    enum Poll<T> {
        Ready(T),
        Pending,
    }

    pub struct Join<FutureA, FutureB> {
        a: Option<FutureA>,
        b: Option<FutureB>,
    }

    impl<FutureA, FutureB> SimpleFuture for Join<FutureA, FutureB>
    where
        FutureA: SimpleFuture<Output = ()>,
        FutureB: SimpleFuture<Output = ()>,
    {
        type Output = ();

        fn poll(&mut self, wake: fn()) -> Poll<Self::Output> {
            if let Some(a) = &mut self.a {
                if let Poll::Ready(()) = a.poll(wake) {
                    self.a.take();
                }
            }

            if let Some(b) = &mut self.b {
                if let Poll::Ready(()) = b.poll(wake) {
                    self.b.take();
                }
            }

            if self.a.is_none() && self.b.is_none() {
                Poll::Ready(())
            } else {
                Poll::Pending
            }
        }
    }

    struct Integer {
        num: i64,
    }

    impl SimpleFuture for Integer {
        type Output = ();

        fn poll(&mut self, wake: fn()) -> Poll<Self::Output> {
            wake();
            Poll::Ready(())
        }
    }

    let mut join = Join {
        a: Some(Integer { num: 1 }),
        b: Some(Integer { num: 2 }),
    };

    println!("{:?}", join.poll(|| println!("hahahaha")))
}

pub fn ch4_21_7() {
    gen_title("ch4_21_7", 18);

    pub struct AndThenFut<FutureA, FutureB> {
        first: Option<FutureA>,
        second: FutureB,
    }

    trait SimpleFuture {
        type Output;
        fn poll(&mut self, wake: fn()) -> Poll<Self::Output>;
    }

    #[derive(Debug)]
    enum Poll<T> {
        Ready(T),
        Pending,
    }

    impl<FutureA, FutureB> SimpleFuture for AndThenFut<FutureA, FutureB>
    where
        FutureA: SimpleFuture<Output = ()>,
        FutureB: SimpleFuture<Output = ()>,
    {
        type Output = ();

        fn poll(&mut self, wake: fn()) -> Poll<Self::Output> {
            if let Some(first) = &mut self.first {
                match first.poll(wake) {
                    Poll::Ready(()) => self.first.take(),
                    Poll::Pending => return Poll::Pending,
                };
            }

            self.second.poll(wake)
        }
    }

    struct Integer {
        num: i64,
    }

    impl SimpleFuture for Integer {
        type Output = ();

        fn poll(&mut self, wake: fn()) -> Poll<Self::Output> {
            wake();
            Poll::Ready(())
        }
    }

    let mut and_the_fut = AndThenFut {
        first: Some(Integer { num: 1 }),
        second: Integer { num: 2 },
    };

    println!("{:?}", and_the_fut.poll(|| println!("hahahaha")));
}

pub fn ch4_21_8() {
    gen_title("ch4_21_8", 18);

    pub struct TimerFuture {
        shared_state: Arc<Mutex<SharedState>>,
    }

    struct SharedState {
        completed: bool,
        waker: Option<Waker>,
    }

    impl Future for TimerFuture {
        type Output = ();

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            let mut shared_sate = self.shared_state.lock().unwrap();
            if shared_sate.completed {
                Poll::Ready(())
            } else {
                shared_sate.waker = Some(cx.waker().clone());
                Poll::Pending
            }
        }
    }

    impl TimerFuture {
        pub fn new(duration: Duration) -> Self {
            let shared_state = Arc::new(Mutex::new(SharedState {
                completed: false,
                waker: None,
            }));

            let thread_shared_state = shared_state.clone();
            thread::spawn(move || {
                thread::sleep(duration);
                let mut shared_state = thread_shared_state.lock().unwrap();
                shared_state.completed = true;
                if let Some(waker) = shared_state.waker.take() {
                    waker.wake()
                }
            });

            TimerFuture { shared_state }
        }
    }

    struct Executor {
        ready_queue: Receiver<Arc<Task>>,
    }

    #[derive(Clone)]
    struct Spawner {
        task_sender: SyncSender<Arc<Task>>,
    }

    struct Task {
        future: Mutex<Option<BoxFuture<'static, ()>>>,
        task_sender: SyncSender<Arc<Task>>,
    }

    fn new_executor_and_spawner() -> (Executor, Spawner) {
        const MAX_QUEUED_TASKS: usize = 10_000;
        let (task_sender, ready_queue) = sync_channel(MAX_QUEUED_TASKS);
        (Executor { ready_queue }, Spawner { task_sender })
    }

    impl Spawner {
        fn spawn(&self, future: impl Future<Output = ()> + 'static + Send) {
            let future = future.boxed();
            let task = Arc::new(Task {
                future: Mutex::new(Some(future)),
                task_sender: self.task_sender.clone(),
            });
            self.task_sender.send(task).expect("任务队列已满");
        }
    }

    impl ArcWake for Task {
        fn wake_by_ref(arc_self: &Arc<Self>) {
            let cloned = arc_self.clone();
            arc_self.task_sender.send(cloned).expect("任务队列已满");
        }
    }

    impl Executor {
        fn run(&self) {
            while let Ok(task) = self.ready_queue.recv() {
                let mut future_slot = task.future.lock().unwrap();
                if let Some(mut future) = future_slot.take() {
                    let waker = waker_ref(&task);
                    let context = &mut Context::from_waker(&*waker);
                    if future.as_mut().poll(context).is_pending() {
                        *future_slot = Some(future);
                    }
                }
            }
        }
    }

    let (executor, spawner) = new_executor_and_spawner();

    spawner.spawn(async {
        println!("howdy!");
        TimerFuture::new(Duration::new(2, 0)).await;
        println!("done!");
    });

    drop(spawner);

    executor.run();
}
